package com.ccc.mq.single.send;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQLocalRequestCallback;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

/**
 * 中心级MQ服务器——消息生产者(发送消息)
 *
 * @author superC
 * @version 1.0.0
 * @date 2022/7/6
 */

@Component
public class SysCenterMQSend {

    protected static final Logger logger = LoggerFactory.getLogger(SysCenterMQSend.class);

    /**
     * 中心MQ
     */
    @Resource(name = "mpCenterMQTemplate")
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 超时时间，单位秒
     */
    @Value("${base.config.rocketmq.timeout:5}")
    private int golbalTimeout;


    /**
     * 单向模式发送
     *
     * @param topic   主题（formats: `topicName` / `topicName:tags`）
     * @param payload 消息数据
     * @date 2022/7/7 9:11
     * @author superC
     */
    public void convertAndSend(String topic, Message<?> payload) {
        logger.info("=======******* 单向模式发送 *******========");
        rocketMQTemplate.convertAndSend(topic, payload);
    }

    /**
     * 单向模式发送
     *
     * @param topic   主题（formats: `topicName` / `topicName:tags`）
     * @param payload 消息数据
     * @param headers the headers for the message to send
     * @date 2022/7/7 9:11
     * @author superC
     */
    public void convertAndSend(String topic, Object payload, Map<String, Object> headers) {
        logger.info("=======******* 单向模式发送 *******========");
        rocketMQTemplate.convertAndSend(topic, payload, headers);
    }

    /**
     * 同步模式发送，并接收消费者的回复
     *
     * @param topic   主题（formats: `topicName` / `topicName:tags`）
     * @param payload 消息数据
     * @date 2022/6/30 15:54
     * @author superC
     */
    public void sendAndReceive(String topic, Object payload) {
        sendAndReceive(topic, payload, golbalTimeout);
    }

    /**
     * 同步模式发送，并接收消费者的回复
     *
     * @param topic   主题（formats: `topicName` / `topicName:tags`）
     * @param payload 消息数据
     * @param timeout 超时时间，单位秒
     * @date 2022/7/6 17:14
     * @author superC
     */
    public void sendAndReceive(String topic, Object payload, int timeout) {
        logger.info("=======******* 同步模式发送 *******========");
        rocketMQTemplate.sendAndReceive(topic, payload, new RocketMQLocalRequestCallback<String>() {
            @Override
            public void onSuccess(String message) {
                logger.info("send user object and receive --- {}", message);
            }

            @Override
            public void onException(Throwable e) {
                logger.error("MQ发送异常！ ---当前消息主题为 {}", topic);
                logger.error("【MQ发送异常】\r\n异常记录：", e);
                // TODO 入库操作，记录同步失败
            }
        }, timeout * 1000);
    }

    /**
     * 同步模式发送顺序消息
     *
     * @param topic   主题（formats: `topicName` / `topicName:tags`）
     * @param payload 消息数据
     * @param hashKey use this key to select queue(MessageQueue). for example: orderId, productId ...
     * @return
     */
    public boolean syncSendOrderly(String topic, Object payload, String hashKey) {
        SendResult sendResult = syncSendOrderly(topic, payload, hashKey, golbalTimeout);
        if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            logger.info("MQ发送成功！ msgId --- {} ， queueOffset --- {}", sendResult.getMsgId(), sendResult.getQueueOffset());
            return true;
        } else {
            logger.error("MQ发送异常！ ---当前消息主题为 {}，异常码：{}", topic, sendResult.getSendStatus());
            return false;
        }
    }

    /**
     * 同步模式发送顺序消息
     *
     * @param topic   主题（formats: `topicName` / `topicName:tags`）
     * @param payload 消息数据
     * @param hashKey use this key to select queue(MessageQueue). for example: orderId, productId ...
     * @param timeout 超时时间，单位秒
     * @return
     */
    public SendResult syncSendOrderly(String topic, Object payload, String hashKey, long timeout) {
        return rocketMQTemplate.syncSendOrderly(topic, payload, hashKey, timeout * 1000);
    }

    public SendResult syncSendOrderly(String topic, Message<?> payload, String hashKey) {
        return syncSendOrderly(topic, payload, hashKey, golbalTimeout);
    }

    public SendResult syncSendOrderly(String topic, Message<?> payload, String hashKey, long timeout) {
        return rocketMQTemplate.syncSendOrderly(topic, payload, hashKey, timeout * 1000);
    }


    /**
     * 异步模式发送请求，并接收MQ的回复
     *
     * @param topic   主题（formats: `topicName` / `topicName:tags`）
     * @param payload 消息数据
     * @date 2022/7/7 9:10
     * @author superC
     */
    public void asyncSend(String topic, Message<?> payload) {
        asyncSend(topic, payload, golbalTimeout);
    }

    /**
     * 异步模式发送请求，并接收MQ的回复
     *
     * @param topic   主题（formats: `topicName` / `topicName:tags`）
     * @param payload 消息数据
     * @param timeout 超时时间，单位秒
     * @date 2022/7/7 9:10
     * @author superC
     */
    public void asyncSend(String topic, Message<?> payload, long timeout) {
        logger.info("=======******* 异步模式发送 *******========");
        rocketMQTemplate.asyncSend(topic, payload, new SendCallback() {
            @Override
            public void onSuccess(SendResult result) {
                logger.info("MQ发送成功！ --- {}", result.toString());
            }

            @Override
            public void onException(Throwable e) {
                logger.error("MQ发送异常！ ---当前消息主题为 {}", topic);
                logger.error("【MQ发送异常】\r\n异常记录：", e);
                // TODO 入库操作，记录同步失败
            }

        }, timeout * 1000);
    }

}